package com.pig4cloud.pigx.strategy.websocket.app.subscribe;

import com.alibaba.fastjson.JSONObject;
import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator;
import com.pig4cloud.pigx.strategy.dto.StrategyContractConditionDTO;
import com.pig4cloud.pigx.strategy.service.StrategyContractService;
import com.pig4cloud.pigx.strategy.vo.StrategyContractAppVO;
import com.pig4cloud.pigx.strategy.websocket.app.dto.WebSocketMessageDTO;
import com.pig4cloud.pigx.strategy.websocket.app.message.SendMessage;
import com.pig4cloud.pigx.strategy.websocket.app.vo.AbstractTopicVO;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 当前策略
 *
 * @since 2022/12/11 13:11
 */
@Slf4j
@Component
public class StrategyContractList extends RequestDataAbstractTopic {

    private final StrategyContractService strategyContractService;
    private final SessionKeyGenerator sessionKeyGenerator;
    private final ThreadPoolTaskExecutor websocketThreadPool;
    public static final String API_NAME = "strategy_contract_list";

    public StrategyContractList(StrategyContractService strategyContractService,
                                SessionKeyGenerator sessionKeyGenerator,
                                ThreadPoolTaskExecutor websocketThreadPool) {
        this.strategyContractService = strategyContractService;
        this.sessionKeyGenerator = sessionKeyGenerator;
        this.websocketThreadPool = websocketThreadPool;
        init();
    }

    private void init() {
        wsPush();
    }

    private void wsPush() {

        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleAtFixedRate(() -> {
            if (subList.isEmpty()) {
                return;
            }

            Set<Map.Entry<WebSocketSession, AbstractTopicVO>> entrySet = subList.entrySet();
            entrySet.forEach(this::push);
        }, 0, 5000, TimeUnit.MILLISECONDS);

    }

    /**
     * 推送数据
     *
     * @param entry 参数
     * @since 2022/12/11 14:25
     */
    @Override
    public void push(Map.Entry<WebSocketSession, AbstractTopicVO> entry) {
        try {
            if (!entry.getKey().isOpen()) {
                subList.remove(entry.getKey());
                return;
            }

            if (startRun(entry.getValue())) {
                // 开启线程，发送数据
                websocketThreadPool.execute(new Task(entry));
            }

        } catch (Throwable e) {
            e.printStackTrace();
            unSub(entry.getKey());
            log.error("webSocket---strategy_contract_list---策略列表Error");
        }
    }

    /**
     * 推送数据
     *
     * @since 2022/9/27 20:23
     */
    @Data
    @AllArgsConstructor
    private class Task implements Runnable {

        private final Map.Entry<WebSocketSession, AbstractTopicVO> entry;

        /**
         * 推送用户合约数据
         *
         * @since 2022/9/27 20:23
         */
        @Override
        public void run() {
            final WebSocketSession session = entry.getKey();
            final AbstractTopicVO param = entry.getValue();

            try {
                // 获取用户id
                Object userId = sessionKeyGenerator.sessionKey(session);
                // 获取参数
                WebSocketMessageDTO dto = entry.getValue().getData();
                JSONObject params = dto.getRequestParam();
                Long groupId = params.getLong("groupId");

                // 组装查询条件
                StrategyContractConditionDTO condition = new StrategyContractConditionDTO();
                condition.setGroupId(groupId);
                condition.setUserId(Long.valueOf(userId.toString()));
                List<StrategyContractAppVO> voList = strategyContractService.appList(condition);

                // 发送消息
                session.sendMessage(SendMessage.getTextMessage(voList));

            } catch (Throwable e) {
                e.printStackTrace();
                log.error("webSocket---strategy_contract_current_detail---当前策略Error");
            } finally {
                endRun(param);
            }
        }
    }

}
